package com.lifq.flink.table

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, dataStreamConversions}
import org.apache.kafka.clients.consumer.OffsetResetStrategy
import org.apache.flink.api.common.eventtime.WatermarkStrategy
/**
 * @Author: LiFuqiang
 * @Date: 2022/8/3 16:23
 * @Version 1.0 
 * @description
 */
object KafkToParquetApp {

  def main(args: Array[String]): Unit = {

    // set up the Scala DataStream API
    //val env = StreamExecutionEnvironment.getExecutionEnvironment
    val env = StreamExecutionEnvironment.createLocalEnvironment(2)
    // set up the Scala Table API
    val tableEnv = StreamTableEnvironment.create(env)

    val ddlSql = "CREATE TABLE MyUserTable (\n  -- declare the schema of the table\n  `user` BIGINT,\n  `message` STRING,\n  `rowtime` TIMESTAMP(3) METADATA FROM 'timestamp',    -- use a metadata column to access Kafka's record timestamp\n  `proctime` AS PROCTIME(),    -- use a computed column to define a proctime attribute\n  WATERMARK FOR `rowtime` AS `rowtime` - INTERVAL '5' SECOND    -- use a WATERMARK statement to define a rowtime attribute\n) WITH (\n  -- declare the external system to connect to\n  'connector' = 'kafka',\n  'topic' = 'topic_name',\n  'scan.startup.mode' = 'earliest-offset',\n  'properties.bootstrap.servers' = 'localhost:9092',\n  'format' = 'json'   -- declare a format for this system\n)"
    tableEnv.executeSql(ddlSql)

    tableEnv.executeSql("select * from MyUserTable")

    tableEnv.useCatalog("");
    tableEnv.useDatabase("")
    val brokers = ""
    val kafkaSource = KafkaSource.builder[String]()
      .setBootstrapServers(brokers)
      .setTopics("input-topic")
      .setGroupId("my-group")
      .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build();


    env.execute("test")
  }
}
